/**
 *
 */
package com.cmcc.pcap.packet.sctp;

import com.cmcc.pcap.packet.tcp.TcpConnection;
import com.cmcc.pcap.packet.tcp.TcpPacketDecorator;
import com.cmcc.pcap.packet.tcp.TcpProtocol;
import com.cmcc.pcap.packet.tcp.TcpProtocolFilter;
import com.cmcc.pcap.packet.util.PacketAnalysisConfig;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.pcap4j.packet.SctpPacket;
import org.pcap4j.packet.TcpPacket;
import org.pcap4j.packet.namednumber.SctpChunkType;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Date 2018年1月17日 上午8:38:12
 * @author Amish
  * 保存一组Connection及其对应的Packet，以connectionKey区分
 * TODO 统计信息单独放到TcpConnectionCacheStatistics类中
 */
public class SctpConnectionCache {
	public static Logger logger = LogManager.getLogger(SctpConnectionCache.class);
	public AtomicInteger totalSctpPacketNum;//缓冲区收到的所有TCP Packet数量
	public AtomicInteger receiveInvalidSctpPacketNum;//收到的所有无效TCP Packet数量（无对应TCP初始消息的）//TODO 无效协议的应该也放到这里？
	public AtomicInteger overTimeInvalidSctpPacketNum;//超时无效并remove掉的TCP Packet数量 （无对应TCP结束消息的）
	public AtomicInteger pickupPacketNum;//缓冲区已经提取出的TCP Packet数量
	public AtomicInteger handledPacketNum;//被工作线程/消费者线程 “处理完”的TCP Packet数量

	public AtomicInteger totalConnectionNum;//进入到cache的Connection总数
	public AtomicInteger finishConnectionNum;//处理完的Connection总数，包括正常处理的、没有结束消息删除的和无效协议删除的等

	public ConcurrentHashMap<String , SctpConnection> chm;//TODO 是否考虑换成性能更好的数据结构
//	public LinkedHashMap <String ,TcpConnection> chm;//用LinkedHashMap替换ConcurrentHashMap，保证遍历的顺序性

	long overtime = 25000;//TCP连接处理的超时时间，超过该时间没有收到FIN/RST包，该连接认为是非法的，直接删除Connection
	long delaytime = 5000;//TCP连接处理的延迟时间，收到FIN或RST后，延迟一定时间处理，以保证后续相关包能放入Connection
	int addthreshold = 20;//向TcpConnectionCache增加pakcet时，生产者阻塞的条件，这里的值表示cache中TcpConnection的数量
	int getthreshold = 6;//从TcpConnectionCache读取TcaConnection时，消费者阻塞的条件，这里的值表示已经读取的符合条件的TcpConnection的数量
	int sctpPacketsThreshold = 10000;//超过该值的tcp connection不做后续处理

	public SctpConnectionCache(){
		chm = new ConcurrentHashMap <String ,SctpConnection>();
		totalSctpPacketNum = new AtomicInteger(0);
		receiveInvalidSctpPacketNum = new AtomicInteger(0);
		overTimeInvalidSctpPacketNum = new AtomicInteger(0);
		pickupPacketNum = new AtomicInteger(0);
		handledPacketNum =  new AtomicInteger(0);
		totalConnectionNum = new AtomicInteger(0);
		finishConnectionNum = new AtomicInteger(0);
		
		overtime = PacketAnalysisConfig.getConfigAsLong("sctp.connection.overtime");
		delaytime = PacketAnalysisConfig.getConfigAsLong("sctp.connection.delaytime");
		addthreshold = PacketAnalysisConfig.getConfigAsInt("sctp.connection.cache.addthreshold");
		getthreshold = PacketAnalysisConfig.getConfigAsInt("sctp.connection.cache.getthreshold");
		sctpPacketsThreshold = PacketAnalysisConfig.getConfigAsInt("sctp.connection.packets.threshold");
	}
	
	/**
	 * 此处会过滤掉没有三次握手消息第一条SYN消息的TcpConnection对应的所有Tcp Packet
	 * @param sctpPacketDecorator
	 */
	public synchronized void addSctpPacketDecorator(SctpPacketDecorator sctpPacketDecorator){
		
		
		this.totalSctpPacketNum.getAndIncrement();
		while( this.chm.size()>=addthreshold ){//TODO 如何更合理的选择这个控制条件？
			try {
				this.wait();
			} catch (InterruptedException e) {
				logger.error(e.getCause(),e);
			}
		}
		//每收到一个TCP包，生成4元组对应的一个Key（注意消息方向），唯一标识一个TCP连接
		String src_ip = sctpPacketDecorator.ipPacket.getHeader().getSrcAddr().getHostAddress();
		String src_port = Integer.toString(sctpPacketDecorator.ipPacket.get(SctpPacket.class).getHeader().getSrcPort().valueAsInt());
		String dst_ip = sctpPacketDecorator.ipPacket.getHeader().getDstAddr().getHostAddress();
		String dst_port =  Integer.toString(sctpPacketDecorator.ipPacket.get(SctpPacket.class).getHeader().getDstPort().valueAsInt());
		String src = src_ip+":"+src_port;
		String dst = dst_ip+":"+dst_port;
		String connectionKey = src+"-"+dst;       //通过四元组生成connectionKey
		String connectionKeyReverse = dst+"-"+src;//通过四元组生成connectionKey，源目的对调

		/**
		 * 如果是初始消息
		 *     *检查集合里是否包含connectionKey对应的TcpConnection对象
		 *         如果没有，新建TcpConnection对象，并保存消息及相关信息
		 *         如果有，并保存消息及相关信息(三次握手的第一条消息也可能重传)
		 * 如果是不是初始消息
		 *     *检查集合里是否包含connectionKey对应的TcpConnection对象
		 *         如果没有对应的TcpConnection，扔掉该消息
		 *         如果有对应的TcpConnection，
		 *             *判断这个是应用层协议是否需要处理
		 *                 需要处理，则保存消息及相关信息
		 *                 不需要处理，直接删除对应的TcpConnection
		 */
		SctpConnection sctpConnection = null;
		try{
			if(sctpPacketDecorator.isNINT){//sctp 四次握手第一条消息，该消息可能有重传消息
				sctpConnection = this.chm.get(connectionKey);

				if(sctpConnection == null ){
					sctpConnection = new SctpConnection();
					sctpConnection.connectionKey = connectionKey;
					sctpConnection.clientAddr = src;
					sctpConnection.serverAddr = dst;
					sctpConnection.al.add(sctpPacketDecorator);
					sctpConnection.lastPacketTime = sctpPacketDecorator.receiveTime;//更新connection最后一个packet的加入cache的时间(注:非抓包时间)
					this.chm.put(connectionKey, sctpConnection);
					this.totalConnectionNum.incrementAndGet();
				}else{
					sctpConnection.al.add(sctpPacketDecorator);
					sctpConnection.lastPacketTime = sctpPacketDecorator.receiveTime;//更新connection最后一个packet的处理时间(注:非抓包时间)
				}
				
			}else{
				sctpConnection = this.chm.get(connectionKey) !=null? this.chm.get(connectionKey) : this.chm.get(connectionKeyReverse);
				
				if(sctpConnection == null){
					receiveInvalidSctpPacketNum.getAndIncrement();
					logger.debug("can not find the match connectionKey with :"+connectionKey+" or "+connectionKeyReverse);
				}else{
					if(sctpConnection.firstDataSctpPacketDecorator == null && sctpPacketDecorator.sctpPacket.getHeader().getChunks().get(0).getType().name().equals(SctpChunkType.DATA.name())){
						/**
						 * 判断应用层协议类型,无法判断时(含HTTP后续分段时)/data长度为零时  都会返回null
						 * 未在配置列表里时会返回NOTDEAL
						 */
						SctpProtocol sctpProtocol = SctpProtocolFilter.filterSctpProtocol(sctpPacketDecorator) ;
						
						if(sctpProtocol != null && sctpProtocol != SctpProtocol.NOTDEAL){
							sctpConnection.firstDataSctpPacketDecorator = sctpPacketDecorator;//给connection设置第一个携带数据的包
							sctpConnection.protocol = sctpProtocol;//给connection设置已识别的应用层协议

							sctpConnection.al.add(sctpPacketDecorator);
							sctpConnection.haveSHUTDOWN = sctpPacketDecorator.chunktypelist.contains(SctpChunkType.SHUTDOWN)? true : false;//标记一个sctpconnection是否有结束报文
							sctpConnection.lastPacketTime = sctpPacketDecorator.receiveTime;//更新connection最后一个packet的处理时间(注:非抓包时间)
						}else{//第一个带data的chunk即无法识别的Protocol，直接删除connection   //TODO 这里忽略了一个包乱序的极小概率事件
							//chm.remove(connectionKey);
							this.chm.remove(sctpConnection.connectionKey);
							this.finishConnectionNum.incrementAndGet();
							this.receiveInvalidSctpPacketNum.getAndAdd(sctpConnection.al.size()+1);//加1是因为当前这个包本身也要属于非法包
						}
					}else{
						sctpConnection.al.add(sctpPacketDecorator);
						sctpConnection.lastPacketTime = sctpPacketDecorator.receiveTime;
						sctpConnection.haveSHUTDOWN = sctpPacketDecorator.chunktypelist.contains(SctpChunkType.SHUTDOWN)? true : false;//标记一个sctpconnection是否有结束报文

					}				
				}
			}		
		
		}catch(Exception e){
			logger.error("Some exception when add packet to cache : ", e);
			if(sctpConnection != null)
			    logger.error(sctpConnection.connectionKey);
			receiveInvalidSctpPacketNum.getAndIncrement();
		}
		this.notify();
	}
	//The web application [/VolteEye_war] appears to have started a thread named [Abandoned connection cleanup thread] but has failed to stop it. This is very likely to create a memory leak.
	
	public synchronized ArrayList<SctpConnection> getSctpConnection(){
		ArrayList<SctpConnection> al = new ArrayList<SctpConnection>();//TODO 传入一个？
		try{
		
			while(chm.values().size() == 0){//TODO 如何选择这个控制条件？
				try {
					this.wait();
				} catch (InterruptedException e) {
					logger.error(e.getCause(),e);
				}
			}
			
			
			int i=getthreshold;//每次处理多少个合法TCP连接？
			
			Iterator<Entry<String,SctpConnection>> iterator = chm.entrySet().iterator();
			while(iterator.hasNext()) {
				if(i==0) {
					this.notify();
					return al;
				}

				SctpConnection sctpConnection = iterator.next().getValue();//entry.getValue();

				if(sctpPacketsThreshold!=0 && sctpConnection.al.size() > sctpPacketsThreshold){//包数超过阈值的链接直接删除20180813
					iterator.remove();//chm.remove(tcpConnection.connectionKey);
					this.finishConnectionNum.incrementAndGet();
					this.receiveInvalidSctpPacketNum.getAndAdd(sctpConnection.al.size());
					logger.info("too long sctp connection,remove this connection and its packet :"+sctpConnection.connectionKey);
					logger.info("remove connection<"+sctpConnection.connectionKey+"> and its packets : "+sctpConnection.al.size());
				}else{
					/*
					if((System.currentTimeMillis() - sctpConnection.al.get(sctpConnection.al.size()-1).receiveTime) > delaytime) {
						al.add(sctpConnection);
						iterator.remove();//chm.remove(tcpConnection.connectionKey);
						this.finishConnectionNum.incrementAndGet();
						pickupPacketNum.getAndAdd(sctpConnection.al.size());

						i--;//放在这里执行是为了保证处理够i设定的数量才释放锁
					}
					 */
					/*
					SCTP协议属于长连接，暂时不考虑超时情况，在没有shutdown报文的情况下，也算一个连接
					 */
					if(!sctpConnection.haveSHUTDOWN){//没有sctp结束事件,sctp连接属于长连接，抓包没有结束事件很正常
						if((System.currentTimeMillis() - sctpConnection.al.get(sctpConnection.al.size()-1).receiveTime) > overtime){//注意，是判断最后一条消息距离当前时间，是不是超过overtime值，TOFIX 这个判断可能与处理性能有关系
							iterator.remove();//chm.remove(tcpConnection.connectionKey);
							this.finishConnectionNum.incrementAndGet();
							overTimeInvalidSctpPacketNum.getAndAdd(sctpConnection.al.size());
							logger.info("overtime and hava not finish or reset message,remove this connection and its packet :"+sctpConnection.connectionKey);
							logger.info("remove connection<"+sctpConnection.connectionKey+"> and its packets : "+sctpConnection.al.size());
							i--;
						}else if ((System.currentTimeMillis() - sctpConnection.al.get(sctpConnection.al.size()-1).receiveTime) > delaytime){//注意，是判断最后一条消息距离当前时间，是不是超过delaytime值，TOFIX 这个判断可能与处理性能有关系
							al.add(sctpConnection);
							iterator.remove();//chm.remove(tcpConnection.connectionKey);
							this.finishConnectionNum.incrementAndGet();
							pickupPacketNum.getAndAdd(sctpConnection.al.size());

							i--;//放在这里执行是为了保证处理够i设定的数量才释放锁
						}else{
							//donothing
						}
					}else {
						if((System.currentTimeMillis() - sctpConnection.al.get(sctpConnection.al.size()-1).receiveTime) > delaytime){//注意，是判断最后一条消息距离当前时间，是不是超过overtime值，TOFIX 这个判断可能与处理性能有关系
							al.add(sctpConnection);
							iterator.remove();//chm.remove(tcpConnection.connectionKey);
							this.finishConnectionNum.incrementAndGet();
							pickupPacketNum.getAndAdd(sctpConnection.al.size());

							i--;//放在这里执行是为了保证处理够i设定的数量才释放锁
						}else {
							//donothing
						}
					}
				}

			}
		
		}catch(Exception e){
			logger.error("Some exception when get packet from cache : ",e);
			//receiveInvalidTcpPacketNum.getAndIncrement();
		}
		this.notify();
		return al;
	}
	
	

}
